package com.dingding.es.util.search;

import co.elastic.clients.elasticsearch.ElasticsearchClient;
import co.elastic.clients.elasticsearch._types.aggregations.Aggregate;
import co.elastic.clients.elasticsearch._types.aggregations.Aggregation;
import co.elastic.clients.elasticsearch._types.query_dsl.Query;
import co.elastic.clients.elasticsearch.cat.aliases.AliasesRecord;
import co.elastic.clients.elasticsearch.cat.indices.IndicesRecord;
import co.elastic.clients.elasticsearch.core.*;
import co.elastic.clients.elasticsearch.core.bulk.BulkResponseItem;
import co.elastic.clients.elasticsearch.core.search.Highlight;
import co.elastic.clients.elasticsearch.core.search.Hit;
import co.elastic.clients.elasticsearch.indices.Alias;
import co.elastic.clients.elasticsearch.indices.CreateIndexRequest;
import co.elastic.clients.util.ObjectBuilder;
import com.alibaba.fastjson.JSONObject;
import com.dingding.es.annotation.DocumentId;
import com.dingding.es.annotation.SearchField;
import com.dingding.es.annotation.TargetClass;
import com.dingding.es.constant.DataType;
import com.google.common.base.Objects;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import org.springframework.util.Assert;
import org.springframework.util.ClassUtils;

import java.io.IOException;
import java.io.Reader;
import java.io.StringReader;
import java.lang.reflect.Field;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.function.Function;
import java.util.stream.Collectors;

/**
 * @author liugongding
 * @description:
 * @date 2023/4/14 4:29 PM
 */

@Slf4j
@Component
public class EsClient {

    private static final String indexPrefix = "index_";
    private static final String aliasPrefix = "alias_";

    @Value("${spring.elasticsearch.rest.alias}")
    private boolean alias;

    @Autowired
    private ElasticsearchClient client;

    //获取类的索引名称(没有指定默认类名首字母小写,  前缀+索引)
    private <T> String getClassIndex(Class<T> clazz){
        TargetClass annotation = clazz.getAnnotation(TargetClass.class);
        String index = annotation.index();
        index = Objects.equal("", index) ? ClassUtils.getShortName(clazz).toLowerCase() : index.toLowerCase();
         return indexPrefix + index;
    }

    //获取类的索引名称(没有指定默认类名首字母小写, 前缀+索引)
    private <T> String getClassAlias(Class<T>  clazz){
        TargetClass annotation = clazz.getAnnotation(TargetClass.class);
        String alias = annotation.alias();
        //反射获取类名
        alias = Objects.equal("",alias) ? ClassUtils.getShortName(clazz).toLowerCase() : alias.toLowerCase();
        return aliasPrefix + alias;
    }

    // 查询全部索引
    public List<String> indexs() {
        List<IndicesRecord> indices = null;
        try {
            indices = client.cat().indices().valueBody();
        } catch (IOException e) {
            e.printStackTrace();
        }
        assert indices != null;
        return indices.stream().map(IndicesRecord::index).collect(Collectors.toList());
    }

    // 查询全部别名
    public List<String> aliases() {
        List<AliasesRecord> aliasesRecords = null;
        try {
            aliasesRecords = client.cat().aliases().valueBody();
        } catch (IOException e) {
            e.printStackTrace();
        }
        assert aliasesRecords != null;
        return aliasesRecords.stream().map(AliasesRecord::alias).collect(Collectors.toList());
    }

    /**
     * 使用索引还是使用别名
     * @param clazz
     * @param <T>
     * @return
     */
    private <T> String getClassAlalsOrIndex(Class<T> clazz){
        if(alias){
            return getClassAlias(clazz);
        }
        return getClassIndex(clazz);

    }

    /**
     * 如果索引或者别名已经存在,那么就不会在创建了
     * @return 是否成功
     * @throws Exception
     */
    public <T> boolean createIndexSettingsMappings(Class<T> tClass) throws Exception {
        TargetClass annotation = tClass.getAnnotation(TargetClass.class);
        String index = getClassIndex(tClass);
        String alias = getClassAlias(tClass);

        List<String> indexs = indexs();
        List<String> aliases = aliases();
        if(indexs.contains(index) || aliases.contains(alias)){
            return false;
        }
        int shards = annotation.shards();
        int replicas = annotation.replicas();
        StringBuilder stringBuilder = new StringBuilder("{");
        stringBuilder.append(
                "\"settings\": {\n"
                        + "    \"number_of_shards\": "
                        + shards
                        + ",\n"
                        + "    \"number_of_replicas\": "
                        + replicas
                        + "\n"
                        + "  },");
        stringBuilder.append("\"mappings\": {\n" + "    \"properties\": ");
        JSONObject jsonObject = new JSONObject();
        //把User的每个属性字段被@SearchField修饰，都把@SearchField里面的字段都添加到JSONObject
        for (Field declaredField : tClass.getDeclaredFields()) {
            //tClass.getDeclaredFields() 获取被类的所有字段
            //取消对象私有变量的安全检查
            declaredField.setAccessible(true);
            //注解上面的各个字段
            JSONObject fielAnnotationdObject = new JSONObject();
            //获取每个字段的被@DocumentId修饰的注解，被该注解标记的字段就是ES的文档id，是唯一的
            DocumentId documentIdAnnotation = declaredField.getAnnotation(DocumentId.class);
            if (documentIdAnnotation != null) {
                fielAnnotationdObject.put("type", DataType.LONG.getType());
                jsonObject.put(declaredField.getName(), fielAnnotationdObject);
                continue;
            }

            SearchField searchFieldAnnotation = declaredField.getAnnotation(SearchField.class);
            if (searchFieldAnnotation != null) {
                //String name = searchFieldAnnotation.name();
                String name = Objects.equal("", searchFieldAnnotation.name()) ? declaredField.getName() : searchFieldAnnotation.name();
                DataType type = searchFieldAnnotation.type();
                String analyzer = searchFieldAnnotation.analyzer();
                String searchAnalyzer = searchFieldAnnotation.searchAnalyzer();
                fielAnnotationdObject.put("type", type.getType());
                if (!Objects.equal("", analyzer)) {
                    fielAnnotationdObject.put("analyzer", analyzer);
                }
                if (!Objects.equal("", searchAnalyzer)) {
                    fielAnnotationdObject.put("search_analyzer", searchAnalyzer);
                }
                //put name,age,description - 但如果被@SearchField的name修饰，key为name的值
                jsonObject.put(name, fielAnnotationdObject);
            }
        }
        Assert.isTrue(jsonObject.size() > 0, "请添加es相关注解");
        stringBuilder.append(jsonObject);
        stringBuilder.append("}}");
        Reader queryJson = new StringReader(stringBuilder.toString());
        final String finalIndex = index;
        final String finalAlias = alias;
        CreateIndexRequest req =
                CreateIndexRequest.of(
                        b ->
                                b.index(finalIndex)
                                        .aliases(finalAlias, Alias.of(a -> a.isWriteIndex(true)))
                                        .withJson(queryJson));
        return client.indices().create(req).acknowledged();
    }

    //查询全部数据
    public <T> List<T> queryAll(Class<T> tClass) throws IOException {
        List<T> list = Lists.newArrayList();
        String index = getClassAlalsOrIndex(tClass);
        SearchResponse<T> search = client.search(s -> s.index(index).query(q -> q.matchAll(m -> m)), tClass);
        //for (Hit<T> hit : search.hits().hits()) {
        //    list.add(hit.source());
        //}
        return search.hits().hits().stream().map(Hit::source).collect(Collectors.toList());
    }


    /**
     * 添加数据
     * @param source 数据源
     * @param async 是否异步，如果是异步那么永远返回null
     * @return
     */
    public <T> String increaseData(T source, boolean async) {
        Object id = null;
        Class<?> tClass = source.getClass();
        for (Field declaredField : tClass.getDeclaredFields()) {
            declaredField.setAccessible(true);
            DocumentId fieldAnnotation = declaredField.getAnnotation(DocumentId.class);
            //找到唯一键的文档id
            if (fieldAnnotation != null) {
                try {
                    //根据对象字段反查对象的值
                    id =  declaredField.get(source);
                } catch (IllegalAccessException e) {
                    e.printStackTrace();
                }
                if (id != null) {
                    break;
                }
            }
        }
        if (id == null) {
            //这里不建议用UUID作为文档id，因为后续无法找到该文档id对其文档进行修改
            //建议文档对象的唯一键不能为空
            id = UUID.randomUUID().toString();
        }
        IndexResponse response = null;
        try {
            IndexRequest.Builder<T> indexReqBuilder = new IndexRequest.Builder<>();
            String index = getClassIndex(source.getClass());
            indexReqBuilder.index(index);
            indexReqBuilder.id(String.valueOf(id));
            indexReqBuilder.document(source);
            //if (async) {
                //异步调用没有返回结果
                //asyncClient.index(indexReqBuilder.build());
                //return null;
            //}
            response = client.index(indexReqBuilder.build());
        } catch (IOException e) {
            e.printStackTrace();
        }
        assert response != null;
        return response.id();
    }

    /**
     * 查询指定文档id数据是否存在
     * @param tClass 索引对应的Class
     * @param id 文档id
     * @param <T> Class对象
     * @return
     * @throws IOException
     */
    public <T> boolean docIdexists(Class<T> tClass, String id) throws IOException {
        //获取索引
        String index = getClassIndex(tClass);
        return client.exists(s -> s.index(index).id(id)).value();
    }

    /**
     * 批量添加数据
     * @param list 需要添加的数据
     * @param async 是否异步
     * @param <T> Class对象
     */
    public <T> void increaseDataBatch(List<T> list, boolean async) {
        BulkRequest.Builder bulk = new BulkRequest.Builder();
        for (T source : list) {
            Object id = null;
            for (Field declaredField : source.getClass().getDeclaredFields()) {
                declaredField.setAccessible(true);
                DocumentId annotation = declaredField.getAnnotation(DocumentId.class);
                if (annotation != null) {
                    try {
                        id =  declaredField.get(source);
                    } catch (IllegalAccessException e) {
                        e.printStackTrace();
                    }
                    if (id != null) {
                        break;
                    }
                }
            }
            if (id == null) {
                id = UUID.randomUUID().getMostSignificantBits();
            }
            Object finalId = id;
            bulk.operations(
                    op -> op.index(
                            idx -> idx.index(getClassAlias(source.getClass()))
                                    .id(String.valueOf(finalId))
                                    .document(source)
                            )
            );
        }

        //if (async) {
        //    asyncClient.bulk(br.build());
        //    return;
        //}
        BulkResponse result = null;
        try {
            result = client.bulk(bulk.build());
        } catch (IOException e) {
            e.printStackTrace();
        }
        // Log errors, if any
        assert result != null;
        if (result.errors()) {
            log.error("Bulk have errors");
            for (BulkResponseItem item : result.items()) {
                if (item.error() != null) {
                    log.error(item.error().reason());
                }
            }
        }
    }

    /**
     * 根据query条件查询
     * @param query 查询条件
     * @param clazz
     * @param <T>
     * @return
     * @throws IOException
     */
    public <T> List<T> complexQuery(Query query, Class<T> clazz) throws IOException {
        List<T> list = Lists.newArrayList();
        String index = getClassIndex(clazz);
        SearchResponse<T> response =
                client.search(s -> s.index(index).query(query), clazz);
        List<Hit<T>> hits = response.hits().hits();
        for (Hit<T> hit : hits) {
            list.add(hit.source());
        }
        return list;
    }

    /**
     * 聚合查询
     * @param query
     * @param function
     * @param clazz
     * @param <T>
     * @return
     * @throws IOException
     */
    public <T> Aggregate complexQueryAggregations(Query query,
                                                  Function<Aggregation.Builder,
                                                  ObjectBuilder<Aggregation>> function,
                                                  Class<T> clazz)throws IOException {

        SearchResponse<T> response =
                client.search(
                        s -> s.index(getClassAlias(clazz))
                                        .size(0) // 不需要显示数据 ,只想要聚合结果
                                        .query(query)
                                        .aggregations("aggregations", function),clazz);

        return response.aggregations().get("aggregations");
    }

    /**
     * 高亮查询
     * @param query
     * @param clazz
     * @param fields
     * @param <T>
     * @return
     * @throws IOException
     */
    public <T> List<Map<String ,Object>> complexQueryHighlight(Query query, Class<T> clazz, String... fields) throws IOException {
        List<Map<String ,Object>> list = Lists.newArrayList();
        String index = getClassAlalsOrIndex(clazz);
        Highlight of = Highlight.of(h -> {
            for (String field : fields) {
                h.fields(field,h1 -> h1.preTags("<font color='red'>").postTags("</font>"));
            }
            return h;
        });
        SearchResponse<T> response = client.search(s -> s
                        .index(index)
                        .query(query)
                        .highlight(of), clazz);
        for (Hit<T> hit : response.hits().hits()) {
            Map<String ,Object> map = Maps.newHashMap();
            map.put("source",hit.source());
            map.put("highlight",hit.highlight());
            list.add(map);
        }
        return list;
    }

    /**
     * 根据文档id查询
     * @param DocumentId
     * @param clazz
     * @param <T>
     * @return
     */
    public <T> T getByDocumentId(String DocumentId, Class<T> clazz) {
        GetResponse<T> response = null;
        try {
            response = client.get(g -> g.index(getClassAlalsOrIndex(clazz)).id(DocumentId), clazz);
        } catch (IOException e) {
            e.printStackTrace();
        }
        assert response != null;
        return response.source();
    }

    /**
     * 根据文档id删除
     * @param documentId
     * @param clazz
     * @param <T>
     * @throws IOException
     */
    public  <T> void deleteByDocumentId( String documentId,Class<T> clazz) throws IOException {
        String index = getClassAlalsOrIndex(clazz);
        DeleteRequest delete = DeleteRequest.of(d -> d.index(index).id(documentId));
        client.delete(delete);
    }

    /**
     * 根据条件删除
     * @param query 条件
     * @param clazz
     * @param <T>
     * @throws IOException
     */
    public  <T> void deleteCondition(Query query, Class<T> clazz) throws IOException {
        String index = getClassAlalsOrIndex(clazz);
        DeleteByQueryRequest delete = DeleteByQueryRequest.of(d -> d.index(index).query(query));
        client.deleteByQuery(delete);
    }

    /**
     * 根据文档id更新
     * @param documentId
     * @param source 源对象
     * @param async
     * @param <T>
     * @throws Exception
     */
    public <T> void updateByDocumentId(String documentId, T source, boolean async) throws Exception {
        //先查询出来被更新的对象
        Object target = getByDocumentId(documentId, source.getClass());
        if (source == null) {
            throw new Exception("没有documentId:"+documentId+"这条数据");
        }
        //进行对象拷贝
        //ObjectCopyUtil.copy(source, docId1);
        BeanUtils.copyProperties(source, target);
        String index = getClassAlalsOrIndex(source.getClass());
        //if (async) {
        //    asyncClient.update(UpdateRequest.of(d -> d.index(index).id(docId).doc(o)), o.getClass());
        //    return;
        //}
        client.update(UpdateRequest.of(d -> d.index(index).id(documentId).doc(source)), source.getClass());
    }

    /**
     * 根据条件更新, 暂时有问题
     * @param query
     * @param source 源对象
     * @param async
     * @param <T>
     * @throws Exception
     */
    public  <T> void updateQuery(Query query, T source, boolean async) throws Exception {
        Class<?> tClass = source.getClass();
        String index = getClassAlalsOrIndex(tClass);
        //获取全部字段和字段的名称以及需要修改的值
        StringBuilder stringBuilder=new StringBuilder();
        for (Field declaredField : tClass.getDeclaredFields()) {
            declaredField.setAccessible(true);
            Object o1 = declaredField.get(source);
            if (o1==null) {
                continue;
            }
            declaredField.setAccessible(true);
            SearchField field = declaredField.getAnnotation(SearchField.class);
            String name = field.name();
            name = "".equals(name) ? declaredField.getName() : name;
            String finalName = name;
            String str = "";
            //String str = PatternCommon.renderString("ctx._source['${name}'] = '${value}';", new HashMap() {{
            //    put("name", finalName);
            //    put("value",o1) ;
            //}});
            stringBuilder.append(str);
        }
        UpdateByQueryRequest update = UpdateByQueryRequest.of(u -> u.index(index).query(query).script(s -> s.inline(i -> i.source(stringBuilder.toString()))));
        //if (async) {
        //    asyncClient.updateByQuery(of);
        //    return;
        //}
        client.updateByQuery(update);
    }

    /**
     * 获取对象的文档id
     * @param source
     * @param <T>
     * @return
     */
    public <T> Object getObjectDocumentId(T source){
        Object id = null;
        for (Field declaredField : source.getClass().getDeclaredFields()) {
            declaredField.setAccessible(true);
            DocumentId annotation = declaredField.getAnnotation(DocumentId.class);
            if (annotation != null) {
                try {
                    id = declaredField.get(source);
                } catch (IllegalAccessException e) {
                    e.printStackTrace();
                }
            }
        }
        return  id;
    }

}
